package com.ibm.cps.processors;

import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import org.apache.log4j.Logger;

import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.ibm.cps.implementations.RuleState;
import com.ibm.cps.message.MessageFields;
import com.ibm.cps.message.OptimizedParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessage;
import com.ibm.cps.message.ParsedDataMessageBuilder;
import com.ibm.cps.newmessage.AbstractMetadata;
import com.ibm.cps.newmessage.RuleEventType;
import com.ibm.cps.newmessage.RuleMetaData;
import com.ibm.cps.redis.RuleStateRedis;
import com.ibm.cps.sql.expression.function.meta.IgnoredInFunctionCall;
import com.ibm.cps.sql.expression.function.meta.ValueParam;
import com.ibm.datamodel.timeseries.TimeSeriesSchema;
import com.ibm.interfaces.IRuleEngine;
import com.ibm.interfaces.ITimeSeriesSchema;
import com.ibm.rules.RuleEngine;
import com.ibm.util.ErrorCode;
import com.ibm.util.JSONUtility;
import com.ibm.util.LocalConfig;
import com.ibm.util.RandomUtil;
import com.ibm.util.exception.CPSException;

// todo
@SuppressWarnings("rawtypes")
@Info(Name=MessageFields.RULE_TYPE,Type=Info.ProcessorType.Other)
public class RuleEngineProcessor extends OptimizedProcessor {

	private static final Logger logger = Logger.getLogger(RuleEngineProcessor.class);
	private IRuleEngine ruleEngine;
	private Map<String, RuleMetaData> id2Rule;
	private Set<String> eventRules;
	private Set<String> filterRules;
	public static RuleStateRedis ruleStateRedis;
	private HashMap<String, RuleState> ruleState;
	private ITimeSeriesSchema parentSchema;
	private ITimeSeriesSchema schema;
	private String metadataId;

	@ValueParam @IgnoredInFunctionCall
	private String ruleName;
	@ValueParam @IgnoredInFunctionCall
	private String ruleId;
	@ValueParam @IgnoredInFunctionCall
	private String ruleString;
	@ValueParam @IgnoredInFunctionCall
	private String isEvent;
	@ValueParam @IgnoredInFunctionCall
	private String parentprocessors;//parentprocessors


//	String ruleName = metaData.getKey2value().get("ruleName");
//	String ruleid = metaData.getKey2value().get("ruleId");
//	String ruleString = metaData.getKey2value().get("ruleString");
//	String isEvent = metaData.getKey2value().get("isEvent");
//	String parentProcessor = metaData.getKey2value().get("parentProcessors");
//	String ruleName = metaData.getKey2value().get(MessageFields.RULENAME);
//	String ruleid = metaData.getKey2value().get(MessageFields.PROCESSORID);
//	String ruleString = metaData.getKey2value().get(MessageFields.RULE);
//	String isEvent = metaData.getKey2value().get(MessageFields.MODEL);
//	String parentProcessor = metaData.getKey2value().get(MessageFields.PARENTPROCESSORS);

	public RuleEngineProcessor(String tenantId, String processorId, String source, ITimeSeriesSchema parentSchema) throws CPSException {
		super(tenantId, processorId, source, parentSchema);
		this.parentSchema = parentSchema;
//		this.processorid = metaMessage.getParentProcessors().iterator().next() + metaMessage.getInputFieldType();
	}

	@Override
	public void open() throws CPSException {
		this.schema = constructSchema();
		this.metadataId = processorId;
		/*
		Edit by SleepingPhoenix
		let processorId like blablablarule instead of ["balblabla"]rule
		 */
//		this.processorId = parentprocessors.substring(2,parentprocessors.length()-2) + MessageFields.RULE_TYPE;
		//this.processorId = parentprocessors + MessageFields.RULE_TYPE;
		// TODO create multipule rule engine?

		ruleEngine = new RuleEngine();

		RuleMetaData rule = new RuleMetaData(tenantId, ruleName, ruleString, isEvent, ruleId, metadataId, parentprocessors);
		ruleEngine.addRule(parentSchema, rule.getRuleid(), rule.getRuleString());

		id2Rule = new HashMap<String, RuleMetaData>();
		eventRules = new HashSet<String>();
		filterRules = new HashSet<String>();
		ruleState = new HashMap<String, RuleState>();
		ruleStateRedis = RuleStateRedis.getRuleStateRedis(LocalConfig.REDIS_SERVER_HOST, LocalConfig.REDIS_SERVER_PORT);
		logger.info("Create rule engine processor successfully.");
	}

	@Override
	public Collection<OptimizedParsedDataMessage> executeSub(ParsedDataMessage dataMessage) throws CPSException {
		if (!(dataMessage instanceof ParsedDataMessage)) {
			throw new CPSException(ErrorCode.INVALID_INPUT_ERROR, "Message shoule be parsed datamessage");
		}
		ParsedDataMessage message = (ParsedDataMessage) dataMessage;
		Object[] value = message.getObjectValues();
		logger.info("Rule engine receive message: " + Arrays.toString(value));
		Collection<BigInteger> hitRules = ruleEngine.executeRules(message);
		logger.info("Hit rules size is " + (hitRules == null ? 0 : hitRules.size()));
		Collection<RuleMetaData> retMsgs = new HashSet<RuleMetaData>();
		if (hitRules != null && hitRules.size() >= 1) {
			retMsgs.addAll(processFilterRules(hitRules, message));
		}
		retMsgs.addAll(processEventRules(hitRules, message));

		if (retMsgs.size() >= 1) {
			return outputMessages(retMsgs, message);
		} else {
			return null;
		}
	}

	private Collection<OptimizedParsedDataMessage> outputMessages(Collection<RuleMetaData> sentRules,
			ParsedDataMessage message) throws CPSException {
		Collection<OptimizedParsedDataMessage> messages = new ArrayList<OptimizedParsedDataMessage>();
//		Comparable[] values;
		int i;
		for (RuleMetaData rule : sentRules) {
// rewrite to support array field
//			Comparable[] existedMsg = message.getValues();
//			values = new Comparable[schema.getTagCount()];
//			for (i = 0; i < existedMsg.length; i++) {
//				values[i] = existedMsg[i];
//			}
//			values[i++] = rule.getRuleName();
//			values[i] = rule.isHit()+"";
//			ParsedDataMessage retMsg = schema.buildParsedDataMessage(values);

			Comparable tskey = message.getTsKey();
			Long timestamp = message.getTimestamp();
			ParsedDataMessageBuilder builder = ParsedDataMessageBuilder.createBasicBuilder(schema, tskey, timestamp);

			builder.CopyValue(parentSchema, message);

			builder.append("hitten",rule.isHit()+"," + rule.isContinueHit());
			builder.append(MessageFields.RULENAME,rule.getRuleName());
			ParsedDataMessage retMsg = builder.toParsedDataMessage();

			OptimizedParsedDataMessage optiMessage = new OptimizedParsedDataMessage(retMsg, rule.getProcessorid());
			messages.add(optiMessage);
			logger.info("Send message " + Arrays.toString(retMsg.getObjectValues()) + " to pubish processor.");
		}
		return messages;
	}

	private Collection<RuleMetaData> processEventRules(Collection<BigInteger> hitRules, ParsedDataMessage message)
			throws CPSException {
		boolean previousState, isHit;
		Collection<RuleMetaData> retMsgs = new HashSet<RuleMetaData>();
		BigInteger id;
		RuleMetaData ruleRecord;

		for (String ruleid : eventRules) {
			previousState = ruleState.get(ruleid).readRuleState(message.getTsKey());
			id = new BigInteger(ruleid, 16);
			isHit = hitRules.contains(id);
			ruleRecord = id2Rule.get(ruleid);
			int eventModel = ruleRecord.getRuleEventMode();
			if (isHit) {
				ruleRecord.setHit(true);
			} else {
				ruleRecord.setHit(false);
			}

			//add by guanxinjun
			// t -> t
			if ((previousState && isHit) && eventModel == RuleEventType.MIX) {
				ruleRecord.setContinueHit(true);
			}else {
				ruleRecord.setContinueHit(false);
			}

			if ((previousState || isHit) && eventModel == RuleEventType.MIX) {
				retMsgs.add(ruleRecord);
				// fix
				ruleState.get(ruleid).setRuleState(message.getTsKey(), isHit);
				continue;
			}

			// f->t
			if (!previousState && isHit) {
				if (eventModel == RuleEventType.FT || eventModel == RuleEventType.FTTF) {
					retMsgs.add(ruleRecord);
				}
				ruleState.get(ruleid).setRuleState(message.getTsKey(), true);
			} else if (previousState && !isHit) { // t -> f
				if (eventModel == RuleEventType.TF || eventModel == RuleEventType.FTTF) {
					retMsgs.add(ruleRecord);
				}
				ruleState.get(ruleid).setRuleState(message.getTsKey(), false);
			}
		}
		return retMsgs;
	}

	private Collection<RuleMetaData> processFilterRules(Collection<BigInteger> hitRules, ParsedDataMessage message) {
		Collection<RuleMetaData> retMsgs = new HashSet<RuleMetaData>();
		BigInteger id;
		RuleMetaData ruleRecord;
		for (String ruleid : filterRules) {
			id = new BigInteger(ruleid, 16);
			if (hitRules.contains(id)) {
				ruleRecord = id2Rule.get(ruleid);
				ruleRecord.setHit(true);
				retMsgs.add(ruleRecord);
			}
		}
		return retMsgs;
	}

	public RuleMetaData constructRuleMetadata(AbstractMetadata metaData) {
		String ruleName = metaData.getKey2value().get("ruleName");
		String ruleid = metaData.getKey2value().get("ruleId");
		String ruleString = metaData.getKey2value().get("ruleString");
		String isEvent = metaData.getKey2value().get("isEvent");
		/*let parentProcessor not like "["hello"]"
		* By SleepingPhoenix
		*/
		//String parentProcessor = metaData.getKey2value().get("parentProcessors");
		//parentProcessor=parentProcessor.substring(1,parentProcessor.length()-1);
		String parentProcessor = metaData.getKey2value().get("parentprocessors");
		RuleMetaData rule = new RuleMetaData(metaData.getTenantid(), ruleName, ruleString, isEvent, ruleid, this.metadataId,
				parentProcessor);
		return rule;
	}

	@Override
	public void addSubProcessor(AbstractMetadata metaData) throws CPSException {
		RuleMetaData rule = constructRuleMetadata(metaData);
		String ruleid = rule.getRuleid();
		ruleEngine.addRule(parentSchema, ruleid, rule.getRuleString());
		id2Rule.put(ruleid, rule);
		if (rule.isEventType()) {
			eventRules.add(ruleid);
		} else {
			filterRules.add(ruleid);
		}
		ruleState.put(ruleid, new RuleState(ruleid, ruleStateRedis));
		subProcessors.add(metaData.getProcessorid());
	}

	@Override
	public void deleteSubProcessor(String processorid) throws CPSException {
		ruleEngine.removeRule(parentSchema, processorid);
		subProcessors.remove(processorid);
		dispose();
	}

	@Override
	public void dispose() {

	}

	private ITimeSeriesSchema constructSchema() throws CPSException {
		ArrayNode tags = parentSchema.getTags().deepCopy();
		ObjectNode ruleNameNode = JSONUtility.newObjectNode();
		ruleNameNode.put(MessageFields.NAME, MessageFields.RULENAME);
		ruleNameNode.put(MessageFields.DATATYPE, "string");
		ObjectNode isHitNode = JSONUtility.newObjectNode();
		isHitNode.put(MessageFields.NAME, "hitten");
		isHitNode.put(MessageFields.DATATYPE, "string");
		tags.add(ruleNameNode);
		tags.add(isHitNode);
		String schemaid = RandomUtil.getRandomId();
		return new TimeSeriesSchema(tenantId, schemaid, tags, 1);
	}

	@Override
	public ITimeSeriesSchema getTimeSeriesSchema() {
		return schema;
	}

}
